package com.example.webflux.demo;

import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;

import javax.annotation.PostConstruct;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

@Service
public class LongPollingService {

    DelayQueue<ConsumerSink<String>> monoSinkQueue = new DelayQueue<>();

    @PostConstruct
    public void timeTask() {
        new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                ConsumerSink<String> sink = null;
                try {
                    sink = monoSinkQueue.poll(100, TimeUnit.MILLISECONDS);
                } catch (InterruptedException ignored) {
                    //noinspection ResultOfMethodCallIgnored
                    Thread.interrupted();
                }
                if (sink != null) {
                    sink.monoSink.success("timeout");
                }
            }
        }).start();
    }

    static class ConsumerSink<T> implements Delayed {
        final MonoSink<T> monoSink;
        private final long time;

        ConsumerSink(MonoSink<T> monoSink, Integer timeMillis) {
            this.monoSink = monoSink;
            this.time = System.currentTimeMillis() + (timeMillis != null && timeMillis > 0 ? timeMillis : 0);
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return time - System.currentTimeMillis();
        }

        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.time, ((ConsumerSink<?>) o).time);
        }
    }

    public Mono<String> polling(Integer timeout) {
        return Mono.create(monoSink -> monoSinkQueue.offer(new ConsumerSink<>(monoSink, timeout)));
    }

    public Mono<String> change(String data) {
        return Mono.create(returnSink -> {
            for (ConsumerSink<String> sink : monoSinkQueue) {
                sink.monoSink.success(data);
            }
            returnSink.success("ok");
        });
    }
}
